package com.yeming.flink.practice.window

import com.yeming.flink.practice.source.StationLog
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object TestReduceFunctionByWindow {

  //每隔五秒计算每个基站的日志数据
  def main(args: Array[String]): Unit = {

    //加载环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //读取数据源
    val stream: DataStream[String] = streamEnv.socketTextStream("f1",9999)
    //数据计算,
    val streamLog: DataStream[StationLog] = stream.map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })

    streamLog.map(log => ((log.sid,1)))
      .keyBy(0)
      .timeWindow(Time.seconds(5))
      .reduce((t1,t2) => {
        (t1._1,t1._2+t2._2)
      })
      .print()

    streamEnv.execute()
  }

}
